package com.wyr.admin.mq;

import com.wyr.admin.mq.config.RabbitMessagePostProcessor;
import com.wyr.admin.mq.config.RabbitMqConfig;
import com.wyr.admin.mq.listener.DynamicQueueListener;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author wyr
 * @version 1.0
 * &#064;title DynamicQueueService
 * &#064;description
 * &#064;date 2025/04/01 17:12
 */
@Slf4j
@Service
public class DynamicQueueService {

    @Resource(name = RabbitMqConfig.DEFAULT_CONNECTION_FACTORY)
    private ConnectionFactory connectionFactory;

    @Autowired
    private RabbitAdmin rabbitAdmin;

    @Autowired
    private DynamicQueueListener dynamicQueueListener;


    /**
     * 存储动态创建的监听容器，方便管理
     **/
    private final Map<String, SimpleMessageListenerContainer> containers = new ConcurrentHashMap<>();

    /**
     * 动态绑定队列和监听器
     *
     * @param queueName
     * &#064;description
     * @return: void
     * @author wangyiren
     * &#064;date 2025/4/2 11:16
     */
    public synchronized void bindDynamicQueueToListener(String queueName) {
        Map<String, Object> arguments = new HashMap<>(1);
        arguments.put("x-single-active-consumer", true);
        Queue queue = new Queue(queueName, true, false, true, arguments);
        // 服务器创建队列
        rabbitAdmin.declareQueue(queue);
        log.info("声明动态队列：{}", queueName);

        // 创建监听容器
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setAfterReceivePostProcessors(new RabbitMessagePostProcessor());
        container.setQueues(queue);
        // 指定监听器适配器
        container.setMessageListener(dynamicQueueListener);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setDefaultRequeueRejected(false);

        // 设置并发消费者数量
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(1);

        // 启动监听容器
        container.start();

        // 将容器添加到Map中以便后续管理
        containers.put(queueName, container);
    }


    /**
     * 延迟删除队列
     *
     * @param queueName 队列名称
     * @param millis    延迟时间(秒)
     */
    public synchronized void deleteQueue(String queueName, long millis) {
        rabbitAdmin.getQueueProperties(queueName);
        try {
            Thread.sleep(millis * 1000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        for (Map.Entry<String, SimpleMessageListenerContainer> entry : containers.entrySet()) {
            SimpleMessageListenerContainer container = entry.getValue();
            if (entry.getKey().equals(queueName) && container.isRunning()) {
                container.stop();
            }
        }

        rabbitAdmin.deleteQueue(queueName);
        log.info("删除队列成功:{}", queueName);
    }
}
